package com.atguigu.app.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.DWDStaticConstants;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class TableProcessFunctionDefine extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private Connection connection;
    private final MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private final OutputTag<JSONObject> outputTag;
    private PreparedStatement preparedStatement;

    public TableProcessFunctionDefine(OutputTag<JSONObject> outputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
        this.outputTag = outputTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(DWDStaticConstants.HBASE_PHOENIX_DRIVER);
        connection = DriverManager.getConnection(DWDStaticConstants.HBASE_PHOENIX_SERVER);
    }

    // value{"database":"","tableName":"","after":"","before":"","type":""}
    @Override
    public void processBroadcastElement(String string, Context context, Collector<JSONObject> collector) throws Exception {
        JSONObject jsonObject = JSON.parseObject(string);
        if (!jsonObject.getString("operation").equalsIgnoreCase("DELETE")){
            String after = jsonObject.getString("after");
            TableProcess tableProcess = JSON.parseObject(after, TableProcess.class);

            // create table
            if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
                checkTable(tableProcess);
            }
            BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
            /**Flink 类型
             * CREATE---->insert
             * UPDATE
             * DELETE
             */
            String key = tableProcess.getSourceTable() + "-" + tableProcess.getOperateType();
            //  base_trademark-insert
            broadcastState.put(key, tableProcess);
        }
    }


    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        String key = jsonObject.getString("table") + "-" + jsonObject.getString("operation").toLowerCase();

        TableProcess tableProcess = broadcastState.get(key);

        if (tableProcess != null) {
            JSONObject after = jsonObject.getJSONObject("after");
            // 过滤字段
            filterColumn(after, tableProcess.getSinkColumns());
            // 分流
            jsonObject.put("sinkTable", tableProcess.getSinkTable());
            if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
                readOnlyContext.output(outputTag, jsonObject);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(tableProcess.getSinkType())) {
                collector.collect(jsonObject);
            }
        } else {
            System.out.println("该配置项" + key + "在配置表中不存在。");
        }

    }

    // 按照配置表中的字段过滤，然后才能传给sink
    private void filterColumn(JSONObject after, String sinkColumns) {
        String[] split = sinkColumns.split(",");
        List<String> list = Arrays.asList(split);
        Iterator<String> iterator = after.keySet().iterator();
        while (iterator.hasNext()) {
            String next = iterator.next();
            boolean contains = list.contains(next);
            if (!contains) {
                iterator.remove();
            }
        }
    }


    private void checkTable(TableProcess tableProcess) {

        if (tableProcess.getSinkPk() == null) {
            tableProcess.setSinkPk("id");
        }
        if (tableProcess.getSinkExtend() == null) {
            tableProcess.setSinkExtend("");
        }
        StringBuilder builder = new StringBuilder("create table if not exists ")
                .append(DWDStaticConstants.HBASE_SCHEMA)
                .append(".")
                .append(tableProcess.getSinkTable())
                .append("(");
        String[] field = tableProcess.getSinkColumns().split(",");
        for (int i = 0; i < field.length; i++) {
            if (field[i].equals(tableProcess.getSinkPk())) {
                builder.append(field[i]).append(" varchar primary key ");
            } else {
                builder.append(field[i]).append(" varchar");
            }
            if (i < field.length - 1) {
                builder.append(",");
            }
        }
        // 建表
        builder.append(")").append(tableProcess.getSinkExtend());
        System.out.println("建表语句: " + builder.toString());
        try {
            preparedStatement = connection.prepareStatement(builder.toString());
            preparedStatement.execute();
        } catch (Exception e) {
            throw new RuntimeException(tableProcess.getSinkTable() + "  建表失败！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
        }

    }
}
